////////////////////////////////////////////////////////////////////////////
//
//  CryEngine Source File.
//  Copyright (C), Crytek.
// -------------------------------------------------------------------------
//  File name:   StreamEngine.cpp
//  Created:     27/07/2010 by Timur.
//  Description: Streaming Engine implementation
// -------------------------------------------------------------------------
//  History:
//
////////////////////////////////////////////////////////////////////////////

#include "StdAfx.h"
#include "StreamEngine.h"
#include "StreamReadStream.h"
#include "../DiskProfiler.h"
#include "../System.h"





#define MAX_HEAVY_ASSETS 20

IStreamEngine::SStatistics* g_pStreamingStatistcs = 0;

//////////////////////////////////////////////////////////////////////////
CStreamEngine::CStreamEngine()
{
	m_nTotalBytesUnziped = 0;
	m_nUnzipBandwidth = 0;
	m_nUnzipBandwidthAverage = 0;
	m_nTempBytesUnziped = 0;
	m_nTempAllocatedMemory = 0;
	m_bStreamingStatsPaused = false;
	m_bInputCallback = false;
	m_nBatchMode = 0;
	m_bShutDown = false;
	m_bTempMemOutOfBudget = false;
	m_bUseOpticalDriveThread = g_cvars.sys_streaming_use_optical_drive_thread!=0;
	m_nPausedDataTypesMask = 0;
	m_bStreamDataOnHDD = gEnv->pCryPak->IsInstalledToHDD();

	g_pStreamingStatistcs = &m_Statistics;
	m_Statistics.nPendingReadBytes = 0;
	m_Statistics.nOpenRequestCount = 0;

	m_Statistics.nCurrentAsyncCount = 0;
	m_Statistics.nCurrentDecompressCount = 0;
	m_Statistics.nCurrentFinishedCount = 0;

	ClearStatistics();

	StartThreads();








	// register system listener
	GetISystem()->GetISystemEventDispatcher()->RegisterListener(this);
}

//////////////////////////////////////////////////////////////////////////
// MT: Main thread only
CStreamEngine::~CStreamEngine()
{
	g_pStreamingStatistcs = 0;
	if (gEnv->pInput && m_bInputCallback)
		gEnv->pInput->RemoveEventListener(this);
	Shutdown();
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::BeginReadGroup()
{
	CryInterlockedIncrement(&m_nBatchMode);
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::EndReadGroup()
{
	CryInterlockedDecrement(&m_nBatchMode);
	if (m_pThreadIO[eIOThread_HDD])
	{
		// New requests accomulated untill all Start stream requests are submitted and can be properly sorted.
		m_pThreadIO[eIOThread_HDD]->SignalStartWork(false);
	}
}

//////////////////////////////////////////////////////////////////////////
// Starts asynchronous read from the specified file
// MT: Main thread only
IReadStreamPtr CStreamEngine::StartRead (const EStreamTaskType tSource, const char* szFilePath, IStreamCallback* pCallback, StreamReadParams* pParams)
{
	if (m_bShutDown)
	{
		return 0;
	}
	unsigned nFlags = 0;
	if (pParams)
		nFlags = pParams->nFlags;

	CReadStream_AutoPtr pStream = new CReadStream(this,tSource,szFilePath, pCallback, pParams);
	
	if (!szFilePath)
	{
		// Special case, remove later.
		m_streams.insert( pStream );
		CAsyncIOFileRequest_AutoPtr pFileRequest = pStream->CreateFileRequest();
		CryInterlockedIncrement( &m_Statistics.nCurrentAsyncCount );
		PushRequestToAsyncCallbackThread( pFileRequest );
		return (CReadStream*)pStream;
	}

	if ((1<<(uint32)tSource) & m_nPausedDataTypesMask)
	{
		// This stream is paused.
		m_pausedStreams.push_back(pStream);
		return (CReadStream*)pStream;
	}

	// Register stream and start file request.	
	m_streams.insert( pStream );
	CAsyncIOFileRequest_AutoPtr pFileRequest = pStream->CreateFileRequest();
	StartFileRequest( pFileRequest );

	return (CReadStream*)pStream;
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::ResumePausedStreams()
{
	for (size_t i = 0; i < m_pausedStreams.size(); )
	{
		CReadStream *pStream = (CReadStream*)(IReadStream*)m_pausedStreams[i];
		int nStreamMask = 1 << (uint32)pStream->m_Type;
		if (0 == (nStreamMask & m_nPausedDataTypesMask))
		{
			if (pStream->GetError() == 0) // If was not aborted
			{
				// This stream must be resumed
				m_streams.insert( pStream );
				CAsyncIOFileRequest_AutoPtr pFileRequest = pStream->CreateFileRequest();
				StartFileRequest( pFileRequest );
			}
			m_pausedStreams.erase( m_pausedStreams.begin()+i );
		}
		else
		{
			i++;
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::StartFileRequest( CAsyncIOFileRequest *pFileRequest )
{
	if (m_pThreadIO[eIOThread_HDD])
	{
		bool bStartImmidietly = m_nBatchMode == 0;
		m_pThreadIO[eIOThread_HDD]->AddRequest( pFileRequest,bStartImmidietly );
	}
	else
	{
		assert(0); // No IO thread.
		return; 
	}

	m_Statistics.typeInfo[pFileRequest->m_eType].nTotalStreamingRequestCount++;

	if (g_cvars.sys_streaming_debug == 3)
	{
		const char* const sFileFilter = g_cvars.sys_streaming_debug_filter_file_name->GetString();

		if (!pFileRequest->m_strFileName.empty() && !m_bStreamingStatsPaused)
			if(!sFileFilter || !sFileFilter[0] || strstr(pFileRequest->m_strFileName.c_str(), sFileFilter))
		{
			CryAutoCriticalSection lock(m_csStats);
			m_statsRequestList.insert( m_statsRequestList.begin(),pFileRequest );
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::SignalToStartWork( EIOThread e, bool bForce )
{
	if( (e >=0) && (e < eIOThread_Last) )
	{
		if( m_pThreadIO[e] )
		{
			m_pThreadIO[e]->SignalStartWork(bForce);
		}
	}	
}

//////////////////////////////////////////////////////////////////////////

void UpdateIOThreadStats(
	IStreamEngine::SStatistics::SMediaTypeInfo* pNotInMemoryInfo, 
	IStreamEngine::SStatistics::SMediaTypeInfo* pInMemoryInfo, 
	CStreamingIOThread* pIOThread,
	float fSecSinceLastReset)
{
	if (pNotInMemoryInfo == 0 || pIOThread == 0)
		return;

	// not in memory reading
	pNotInMemoryInfo->fActiveDuringLastSecond = pIOThread->m_NotInMemoryStats.m_fReadingDuringLastSecond;
	pNotInMemoryInfo->fAverageActiveTime = 
		pIOThread->m_NotInMemoryStats.m_TotalReadTime.GetSeconds() / fSecSinceLastReset * 100;

	pNotInMemoryInfo->nBytesRead = pIOThread->m_NotInMemoryStats.m_nReadBytesInLastSecond;
	pNotInMemoryInfo->nRequestCount = pIOThread->m_NotInMemoryStats.m_nRequestCountInLastSecond;
	pNotInMemoryInfo->nTotalBytesRead = pIOThread->m_NotInMemoryStats.m_nTotalReadBytes;
	pNotInMemoryInfo->nTotalRequestCount = pIOThread->m_NotInMemoryStats.m_nTotalRequestCount;

	pNotInMemoryInfo->nSeekOffsetLastSecond = pIOThread->m_NotInMemoryStats.m_nReadOffsetInLastSecond;
	if (pIOThread->m_NotInMemoryStats.m_nTotalRequestCount > 0) {
		pNotInMemoryInfo->nAverageSeekOffset = pIOThread->m_NotInMemoryStats.m_nTotalReadOffset / 
			pIOThread->m_NotInMemoryStats.m_nTotalRequestCount;
	} else {
		pNotInMemoryInfo->nAverageSeekOffset = 0;
	}

	pNotInMemoryInfo->nCurrentReadBandwidth = pIOThread->m_NotInMemoryStats.m_nCurrentReadBandwith;
	pNotInMemoryInfo->nSessionReadBandwidth = (uint32)(pNotInMemoryInfo->nTotalBytesRead / fSecSinceLastReset);

	pNotInMemoryInfo->nActualReadBandwidth = pIOThread->m_NotInMemoryStats.m_nActualReadBandwith;
	float fTotalReadTime = pIOThread->m_NotInMemoryStats.m_TotalReadTime.GetSeconds();
	if (fTotalReadTime > 0.0f)
		pNotInMemoryInfo->nAverageActualReadBandwidth = (uint32)(pNotInMemoryInfo->nTotalBytesRead / fTotalReadTime);

	// in memory reading
	if (pInMemoryInfo)
	{
		pInMemoryInfo->nBytesRead = pIOThread->m_InMemoryStats.m_nReadBytesInLastSecond;
		pInMemoryInfo->nRequestCount = pIOThread->m_InMemoryStats.m_nRequestCountInLastSecond;
		pInMemoryInfo->nTotalBytesRead = pIOThread->m_InMemoryStats.m_nTotalReadBytes;
		pInMemoryInfo->nTotalRequestCount = pIOThread->m_InMemoryStats.m_nTotalRequestCount;
	}
}

void CStreamEngine::Update(uint32 nUpdateTypesBitmask)
{
	FUNCTION_PROFILER( GetISystem(), PROFILE_SYSTEM );
	LOADING_TIME_PROFILE_SECTION(gEnv->pSystem);

	// Dispatch completed callbacks.
	MainThread_FinalizeIOJobs(nUpdateTypesBitmask);

}

// Gets called regularly, to finalize those proxies whose jobs have
// already been executed (e.g. to call the callbacks)
//  - to be called from the main thread only
//  - starts new jobs in the single-threaded model
void CStreamEngine::Update()
{
	FUNCTION_PROFILER( GetISystem(), PROFILE_SYSTEM );
	LOADING_TIME_PROFILE_SECTION(gEnv->pSystem);

	// Dispatch completed callbacks.
	MainThread_FinalizeIOJobs();

	if (g_cvars.sys_streaming_resetstats)
	{
		ClearStatistics();
		g_cvars.sys_streaming_resetstats = 0;
	}

	CTimeValue t = gEnv->pTimer->GetAsyncTime();
	if ((t-m_nLastBandwidthUpdateTime).GetMilliSecondsAsInt64() > 1000)
	{
		// Repeat every second.
		m_nUnzipBandwidth = 0;
		if (m_tempUnzipTime.GetValue() != 0)
		{
			m_nUnzipBandwidth = (uint32)(m_nTempBytesUnziped / m_tempUnzipTime.GetSeconds());
		}
		m_tempUnzipTime.SetValue(0);
		m_nTempBytesUnziped = 0;

		m_nLastBandwidthUpdateTime = t;
	}
	if (m_totalUnzipTime.GetValue() != 0)
	{
		m_nUnzipBandwidthAverage = (uint32)(m_nTotalBytesUnziped / m_totalUnzipTime.GetSeconds());
	}

	m_Statistics.nDecompressBandwidth = m_nUnzipBandwidth;
	m_Statistics.nDecompressBandwidthAverage = m_nUnzipBandwidthAverage;

	CTimeValue currentTime = gEnv->pTimer->GetAsyncTime();

	CTimeValue timeSinceLastReset = currentTime - m_TimeOfLastReset;
	float fSecSinceLastReset = timeSinceLastReset.GetSeconds();

	CTimeValue timeSinceLastUpdate = currentTime - m_TimeOfLastUpdate;

	// update the stats every second
	if (timeSinceLastUpdate.GetMilliSecondsAsInt64() > 1000)
	{
		UpdateIOThreadStats(&m_Statistics.hddInfo, &m_Statistics.memoryInfo, m_pThreadIO[eIOThread_HDD], fSecSinceLastReset);
		UpdateIOThreadStats(&m_Statistics.discInfo, 0, m_pThreadIO[eIOThread_Optical], fSecSinceLastReset);		
		UpdateIOThreadStats(&m_Statistics.memoryInfo, 0, m_pThreadIO[eIOThread_InMemory], fSecSinceLastReset);		

		SStatistics::SRequestTypeInfo totals;

		// update stats on all types
		for (int i = 0; i < eStreamTaskTypeCount; i++) {
			SStatistics::SRequestTypeInfo& info = m_Statistics.typeInfo[i];

			if (info.nTotalStreamingRequestCount)
				info.fAverageCompletionTime = info.fTotalCompletionTime / info.nTotalStreamingRequestCount;
			else
				info.fAverageCompletionTime = 0;
			info.nSessionReadBandwidth = (uint32)(info.nTotalReadBytes / fSecSinceLastReset);
			info.nCurrentReadBandwidth = (uint32)(info.nTmpReadBytes / timeSinceLastUpdate.GetSeconds());

			info.fAverageRequestCount = info.nTotalStreamingRequestCount / fSecSinceLastReset;

			totals.Merge(info);

			info.nTmpReadBytes = 0;
		}

		if (totals.nTotalStreamingRequestCount > 0)
			m_Statistics.fAverageCompletionTime = totals.fTotalCompletionTime / totals.nTotalStreamingRequestCount;

		m_Statistics.nTotalSessionReadBandwidth = (uint32)(totals.nTotalReadBytes / fSecSinceLastReset);
		m_Statistics.nTotalCurrentReadBandwidth = (uint32)(totals.nTmpReadBytes / timeSinceLastUpdate.GetSeconds());
		m_Statistics.fAverageRequestCount = totals.nTotalStreamingRequestCount / fSecSinceLastReset;

		m_Statistics.nTotalRequestCount = totals.nTotalRequestCount;
		m_Statistics.nTotalStreamingRequestCount = totals.nTotalStreamingRequestCount;
		m_Statistics.nTotalBytesRead = totals.nTotalReadBytes;

		// update this flag only once a second to be sure it's visible in display info
		m_Statistics.bTempMemOutOfBudget = m_bTempMemOutOfBudget;
		m_bTempMemOutOfBudget = false;

		m_TimeOfLastUpdate = currentTime;
	}

	int nTmpAllocated = m_nTempAllocatedMemory;
	m_Statistics.nMaxTempMemory = max(m_Statistics.nMaxTempMemory, nTmpAllocated);

	if(m_Statistics.vecHeavyAssets.size() > MAX_HEAVY_ASSETS)
	{
		std::sort(m_Statistics.vecHeavyAssets.begin(), m_Statistics.vecHeavyAssets.end());
		m_Statistics.vecHeavyAssets.resize(MAX_HEAVY_ASSETS);
	}

	// allow changing usage of optical drive thread on the fly for profiling
	if (g_cvars.sys_streaming_use_optical_drive_thread != (int)m_bUseOpticalDriveThread)
	{
		m_bUseOpticalDriveThread = g_cvars.sys_streaming_use_optical_drive_thread!=0;
		if (m_bUseOpticalDriveThread)
			m_pThreadIO[eIOThread_HDD]->RegisterFallbackIOThread(eStreamSourceTypeDisc, m_pThreadIO[eIOThread_Optical]);
	}
	if (!(gEnv->IsDedicated()))
	{
		//register in memory fallback thread
		m_pThreadIO[eIOThread_HDD]->RegisterFallbackIOThread(eStreamSourceTypeMemory, m_pThreadIO[eIOThread_InMemory]);
	}

	if (g_cvars.sys_streaming_debug)
	{
		DrawStatistics();

		if (!m_bInputCallback)
		{
			if (gEnv->pInput)
				gEnv->pInput->AddEventListener(this);
			m_bInputCallback = true;
		}
	}


}

//////////////////////////////////////////////////////////////////////////
// Only waits at most the specified amount of time for some IO to complete
void CStreamEngine::UpdateAndWait(bool bAbortAll)
{
	// for stream->Wait sync
	LOADING_TIME_PROFILE_SECTION(gEnv->pSystem);

	if (bAbortAll)
	{
		for (int i = 0; i < eIOThread_Last; i++)
		{
			if (m_pThreadIO[i])
				m_pThreadIO[i]->AbortAll(bAbortAll);
		}
	}

	while (!m_finishedStreams.empty() || !m_streams.empty())
	{
		Update();
    // In case we still have cancelled or aborted streams in the queue, 
    // we wake the io threads here to ensure they are removed correctly; 
		for (uint32 i = 0; i < (uint32)eIOThread_Last; ++i)
			SignalToStartWork( (EIOThread)i, true );
		CrySleep(10);
	}

	if (bAbortAll)
	{
		for (int i = 0; i < eIOThread_Last; i++)
		{
			if (m_pThreadIO[i])
				m_pThreadIO[i]->AbortAll(false);
		}
	}
}

// In the Multi-Threaded model (with the IO Worker thread)
// removes the proxies from the IO Queue as needed, and the proxies may call their callbacks
void CStreamEngine::MainThread_FinalizeIOJobs(uint32 type)
{
	static bool bNoReentrant = false;

	if (!bNoReentrant)
	{
		bNoReentrant = true;

		FUNCTION_PROFILER( GetISystem(), PROFILE_SYSTEM );
		LOADING_TIME_PROFILE_SECTION(gEnv->pSystem);

		m_Statistics.nMainStreamingThreadWait = CryGetTicks();
		/*
		{
			int nIO = m_pThreadIO[0]->GetRequestCount();
			int nUnzip = m_decompressThreads[0]->GetRequestCount();
			int nCallback = m_asyncCallbackThread->GetRequestCount();
			string str; str.Format( "Stream Requests IO=%d   Unzip=%d   ACall=%d\n",nIO,nUnzip,nCallback ); OutputDebugString(str.c_str());
		}
		*/

		int nCount = 0;

		CryMT::vector<CReadStream_AutoPtr> finishedStreams;
		// Dispatch completed callbacks.
		CReadStream_AutoPtr pStream(0);
		while (m_finishedStreams.try_pop_front( pStream ))
		{
			if (pStream->m_Type & type) 
			{

				CryInterlockedDecrement(&m_Statistics.nCurrentFinishedCount);

				pStream->MainThread_Finalize();
				// update statistics
				UpdateStatistics( pStream );
				m_streams.erase(pStream);


				nCount++;
				// perform time slicing if requested
				if (g_cvars.sys_streaming_max_finalize_per_frame > 0 &&
					nCount > g_cvars.sys_streaming_max_finalize_per_frame)
				{
					break;
				}
			}
			else
			{
				finishedStreams.push_back(pStream);
			}
		}

		bNoReentrant = false; 

		while (finishedStreams.try_pop_front( pStream ))
		{
			m_finishedStreams.push_back(pStream);
		}

		m_Statistics.nMainStreamingThreadWait = CryGetTicks() - m_Statistics.nMainStreamingThreadWait;
	}
}


// In the Multi-Threaded model (with the IO Worker thread)
// removes the proxies from the IO Queue as needed, and the proxies may call their callbacks
void CStreamEngine::MainThread_FinalizeIOJobs()
{
	static bool bNoReentrant = false;

	if (!bNoReentrant)
	{
		bNoReentrant = true;

		FUNCTION_PROFILER( GetISystem(), PROFILE_SYSTEM );
		LOADING_TIME_PROFILE_SECTION(gEnv->pSystem);

		m_Statistics.nMainStreamingThreadWait = CryGetTicks();
		/*
		{
			int nIO = m_pThreadIO[0]->GetRequestCount();
			int nUnzip = m_decompressThreads[0]->GetRequestCount();
			int nCallback = m_asyncCallbackThread->GetRequestCount();
			string str; str.Format( "Stream Requests IO=%d   Unzip=%d   ACall=%d\n",nIO,nUnzip,nCallback ); OutputDebugString(str.c_str());
		}
		*/

		int nCount = 0;

		// Dispatch completed callbacks.
		CReadStream_AutoPtr pStream(0);
		while (m_finishedStreams.try_pop_front( pStream ))
		{
			CryInterlockedDecrement(&m_Statistics.nCurrentFinishedCount);

			pStream->MainThread_Finalize();
			// update statistics
			UpdateStatistics( pStream );
			m_streams.erase(pStream);			

			nCount++;
			// perform time slicing if requested
			if (g_cvars.sys_streaming_max_finalize_per_frame > 0 &&
				nCount > g_cvars.sys_streaming_max_finalize_per_frame)
			{
				break;
			}
		}

		bNoReentrant = false; 

		m_Statistics.nMainStreamingThreadWait = CryGetTicks() - m_Statistics.nMainStreamingThreadWait;
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::UpdateJobPriority(IReadStreamPtr pJobStream)
{
	for (int i = 0; i < eIOThread_Last; i++)
	{
		if (m_pThreadIO[i])
		{
			m_pThreadIO[i]->NeedSorting();
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::StopThreads()
{
	for (int i = 0; i < eIOThread_Last; i++)
	{
		m_pThreadIO[i] = 0;
	}

	m_decompressThreads.clear();
	m_asyncCallbackThreads.clear();
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::StartThreads()
{
	StopThreads();

	m_pThreadIO[eIOThread_HDD] = new CStreamingIOThread(this,eStreamSourceTypeHDD,"Streaming File IO HDD");
	if (!(gEnv->IsDedicated()))
	{
		m_pThreadIO[eIOThread_Optical] = new CStreamingIOThread(this,eStreamSourceTypeDisc,"Streaming File IO Optical");
		m_pThreadIO[eIOThread_InMemory] = new CStreamingIOThread(this,eStreamSourceTypeMemory,"Streaming File IO InMemory", 150/*needs high prio*/);
	}

	if (m_bUseOpticalDriveThread)
	{
		// register optical as fallback type on HDD
		m_pThreadIO[eIOThread_HDD]->RegisterFallbackIOThread(eStreamSourceTypeDisc, m_pThreadIO[eIOThread_Optical]);
	}
	if (!(gEnv->IsDedicated()))
	{
		m_pThreadIO[eIOThread_HDD]->RegisterFallbackIOThread(eStreamSourceTypeMemory, m_pThreadIO[eIOThread_InMemory]);
	}

	// More decompress threads can be added here.
	m_decompressThreads.push_back( new CStreamingWorkerThread(this,"Streaming Unzip 1",CStreamingWorkerThread::eWorkerDecompress) );
	m_asyncCallbackThreads.push_back( new CStreamingWorkerThread(this,"Streaming AsyncCallback",CStreamingWorkerThread::eWorkerAsyncCallback) );





}

//! Puts the memory statistics into the given sizer object
//! According to the specifications in interface ICrySizer
void CStreamEngine::GetMemoryStatistics(ICrySizer *pSizer)
{	
	SIZER_COMPONENT_NAME(pSizer, "CRefStreamEngine");

	size_t nSize = sizeof(*this);

	pSizer->AddObject(this, nSize);
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::AbortJob( CReadStream* pStream )
{	
	if (m_finishedStreams.try_remove( (CReadStream*)pStream ))
		CryInterlockedDecrement(&m_Statistics.nCurrentFinishedCount);

	if(!m_pausedStreams.empty())
	{
		std::vector<CReadStream_AutoPtr>::iterator it = std::find(
			m_pausedStreams.begin(), m_pausedStreams.end(), pStream);
		if(it != m_pausedStreams.end())
			m_pausedStreams.erase(it);
	}

	m_streams.erase( pStream );
}

IStreamEngine::SStatistics& CStreamEngine::GetStreamingStatistics()
{
	return m_Statistics;
}

void CStreamEngine::UpdateStatistics( CReadStream *pReadStream )
{
#if	!defined(_RELEASE) || defined(ENABLE_STATOSCOPE_RELEASE)
	uint32 nBytesRead = pReadStream->m_nBytesRead;

	SStatistics::SRequestTypeInfo& info = m_Statistics.typeInfo[pReadStream->m_Type];
	info.nTotalRequestCount++;

	// only add to stats if request was valid
	const string& name = pReadStream->GetName();
	if (name.length() > 0)
	{
		info.nTotalReadBytes += nBytesRead;
		info.nTmpReadBytes += nBytesRead;
		info.nTotalRequestDataSize += pReadStream->m_Params.nSize;

		CTimeValue completionTime = gEnv->pTimer->GetAsyncTime() - pReadStream->GetRequestTime();
		float fCompletionTime = completionTime.GetMilliSeconds();
		info.fTotalCompletionTime += fCompletionTime;

		size_t splitter = name.find_last_of(".");
		if (splitter != string::npos) {
			string extension = name.substr(splitter + 1);
			TExtensionInfoMap::iterator findRes = m_PerExtensionInfo.find(extension);
			if (findRes == m_PerExtensionInfo.end())
			{
				m_PerExtensionInfo[extension] = SExtensionInfo();
				findRes = m_PerExtensionInfo.find(extension);
			}

			SExtensionInfo& extensionInfo = findRes->second;
			extensionInfo.m_fTotalReadTime += pReadStream->m_ReadTime.GetMilliSeconds();
			extensionInfo.m_nTotalRequests++;
			extensionInfo.m_nTotalReadSize += nBytesRead;
			extensionInfo.m_nTotalRequestSize += pReadStream->m_Params.nSize;
		}
	}	

	if(nBytesRead > 64*1024)
	{
		m_Statistics.vecHeavyAssets.push_back(SStatistics::SAsset(pReadStream->m_strFileName, nBytesRead));
	}
#endif
}

void CStreamEngine::Shutdown()
{
	m_bShutDown = true;
	UpdateAndWait(true);	
	CancelAll();

	StopThreads();

	m_streams.clear();
	m_finishedStreams.clear();

	// unregister system listener
	GetISystem()->GetISystemEventDispatcher()->RemoveListener(this);
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::CancelAll()
{
	for (int i = 0; i < eIOThread_Last; i++)
	{
		if (m_pThreadIO[i])
			m_pThreadIO[i]->BeginReset();
	}

	for (int i = 0; i < eIOThread_Last; i++)
	{
		if (m_pThreadIO[i])
			m_pThreadIO[i]->EndReset();
	}

	for (size_t i = 0; i < m_decompressThreads.size(); ++i)
	{
		m_decompressThreads[i]->BeginReset();
	}
	for (size_t i = 0; i < m_decompressThreads.size(); ++i)
	{
		m_decompressThreads[i]->EndReset();
	}


	for (size_t i = 0; i < m_asyncCallbackThreads.size(); ++i)
	{
		m_asyncCallbackThreads[i]->BeginReset();
	}
	for (size_t i = 0; i < m_asyncCallbackThreads.size(); ++i)
	{
		m_asyncCallbackThreads[i]->EndReset();
	}

	// make sure we don't check for canceled tasks when destroying the m_finishedStreams container	
	m_streams.clear();
	stl::free_container(m_finishedStreams);
	stl::free_container(m_pausedStreams);
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::ReportAsyncFileRequestReadDone( CAsyncIOFileRequest_AutoPtr pFileRequest )
{
	if (!pFileRequest->IsCanceled())
	{
		if (pFileRequest->m_bCompressedBuffer)
		{
			// Need to be decompressed.
			///
			m_decompressThreads[0]->AddRequest( pFileRequest );

			CryInterlockedIncrement(&m_Statistics.nCurrentDecompressCount);
		}
		else
		{
			ReportAsyncFileRequestDecompressed(pFileRequest);
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::ReportAsyncFileRequestDecompressed( CAsyncIOFileRequest_AutoPtr pFileRequest )
{
	if (!pFileRequest->IsCanceled())
	{
		// Update decompression statistics.
		if (pFileRequest->m_bCompressedBuffer)
		{
			CryInterlockedDecrement(&m_Statistics.nCurrentDecompressCount);

			if (pFileRequest->m_unzipTime.GetValue() != 0)
			{
				m_nTotalBytesUnziped += pFileRequest->m_nFileSize;
				m_totalUnzipTime += pFileRequest->m_unzipTime;

				m_nTempBytesUnziped += pFileRequest->m_nFileSize;
				m_tempUnzipTime += pFileRequest->m_unzipTime;
			}
		}
		PushRequestToAsyncCallbackThread(pFileRequest);

		CryInterlockedIncrement(&m_Statistics.nCurrentAsyncCount);
	}
}


//////////////////////////////////////////////////////////////////////////
void CStreamEngine::ReportAsyncFileRequestComplete( CAsyncIOFileRequest_AutoPtr pFileRequest )
{
	if (!pFileRequest->IsCanceled())
	{
		CryInterlockedDecrement(&m_Statistics.nCurrentAsyncCount);

		if (pFileRequest->m_pCallback)
		{
			pFileRequest->m_pCallback->OnAsyncFinished(pFileRequest);
		}
		if (pFileRequest->m_pReadStream)
		{
			CReadStream_AutoPtr pStream = (CReadStream*)(IReadStream*)pFileRequest->m_pReadStream;
			pStream->OnAsyncFileRequestComplete();
			m_finishedStreams.push_back(pStream);

			CryInterlockedIncrement(&m_Statistics.nCurrentFinishedCount);
		}
		pFileRequest->Done();

		if (g_cvars.sys_streaming_debug != 0)
		{
			if (g_cvars.sys_streaming_debug == 2 || g_cvars.sys_streaming_debug == 4)
			{
				const char* const sFileFilter = g_cvars.sys_streaming_debug_filter_file_name->GetString();

				if (!pFileRequest->m_strFileName.empty() && !m_bStreamingStatsPaused)
					if(!sFileFilter || !sFileFilter[0] || strstr(pFileRequest->m_strFileName.c_str(), sFileFilter))
				{
					CryAutoCriticalSection lock(m_csStats);
					m_statsRequestList.insert( m_statsRequestList.begin(),pFileRequest );
				}
			}
		}
	}
}

//////////////////////////////////////////////////////////////////////////
const char* CStreamEngine::GetStreamTaskTypeName( EStreamTaskType type )
{
	switch (type)
	{
	case eStreamTaskTypeMusic:
		return "Music";
	case eStreamTaskTypeAnimation:
		return "Animation";
	case eStreamTaskTypeGeometry:
		return "Geometry";
	case eStreamTaskTypeSound:
		return "Sound";
	case eStreamTaskTypeTexture:
		return "Texture";
	case eStreamTaskTypeShader:
		return "Shader";
	case eStreamTaskTypeTerrain:
		return "Terrain";
	case eStreamTaskTypeVideo:
		return "Video";
	case eStreamTaskTypeFlash:
		return "Flash";
	case eStreamTaskTypeReadAhead:
		return "ReadAhead";
	case eStreamTaskTypePak:
		return "Pak";
	}
	return "";
}

void CStreamEngine::GetBandwidthStats(EStreamTaskType type, float * bandwidth)
{
#if !defined (_RELEASE) || defined(ENABLE_STATOSCOPE_RELEASE)
	*bandwidth = m_Statistics.typeInfo[type].nCurrentReadBandwidth / 1024.0f;
#endif
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::ReportTempMemAlloc( uint32 nSizeAlloc,uint32 nSizeFree )
{
	int nAdd = (int)nSizeAlloc - (int)nSizeFree;
	CryInterlockedAdd( &m_nTempAllocatedMemory,nAdd );
}

namespace
{
	void DrawText( const float x, const float y,ColorF c,const char * format, ... )
	{
		va_list args;
		va_start(args, format);

		SDrawTextInfo ti;
		ti.flags = eDrawText_FixedSize | eDrawText_2D | eDrawText_Monospace;
		ti.xscale = ti.yscale = 1.2f;
		ti.color[0] = c.r;
		ti.color[1] = c.g;
		ti.color[2] = c.b;
		ti.color[3] = c.a;
		gEnv->pRenderer->DrawTextQueued( Vec3(x,y,1.0f),ti,format,args );
		va_end(args);
	}
	void WriteToStreamingLog( const char *str )
	{
#ifndef _RELEASE
		if (g_cvars.sys_streaming_debug == 4)
		{
			// ignore invalid file access when logging steaming data
			CDebugAllowFileAccess ignoreInvalidFileAccess;

			static string sFileName;
			static bool bFirstTime = true;
			if (bFirstTime)
			{
				char path[ICryPak::g_nMaxPath];
				path[sizeof(path) - 1] = 0;
				gEnv->pCryPak->AdjustFileName("TestResults\\StreamingLog.txt", path, ICryPak::FLAGS_PATH_REAL|ICryPak::FLAGS_FOR_WRITING);
				sFileName = path;
			}
			FILE *file = fxopen( sFileName, (bFirstTime) ? "wt" : "at" );
			bFirstTime = false;
			if (file)
			{
				fprintf( file,"%s\n",str );
				fclose(file);
			}
		}
#endif
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::DrawStatistics()
{
	std::vector<CAsyncIOFileRequest_AutoPtr> tempRequests;

	if (g_cvars.sys_streaming_debug == 4)
	{
#ifndef _RELEASE
		float tx = 0;
		float ty = 30;
		float ystep = 12.0f;
		ColorF clText(1,0,0,1);

		//take overscan borders into account
		Vec2 overscanBorders = *(Vec2*)gEnv->pRenderer->EF_Query(EFQ_OverscanBorders);
		tx += ((float)gEnv->pRenderer->GetWidth()) * overscanBorders.x;
		ty += ((float)gEnv->pRenderer->GetHeight()) * overscanBorders.y;

		DrawText( tx,ty+=ystep,clText, "Recording streaming stats to file ..." );

		{
			CryAutoCriticalSection lock(m_csStats);
			tempRequests.swap(m_statsRequestList);
		}

		const char* const sFileFilter = g_cvars.sys_streaming_debug_filter_file_name->GetString();

		if (!tempRequests.empty())
		{
			for(int i = (int)tempRequests.size()-1; i >= 0; i--)
			{
				CAsyncIOFileRequest *pFileRequest = tempRequests[i];

				if (g_cvars.sys_streaming_debug_filter > 0 && pFileRequest->m_eType != g_cvars.sys_streaming_debug_filter)
					continue;
				if (g_cvars.sys_streaming_debug_filter == -1 && pFileRequest->m_eMediaType == eStreamSourceTypeMemory)
					continue;
				if (g_cvars.sys_streaming_debug_filter_min_time && (pFileRequest->m_readTime.GetMilliSeconds() < (float)g_cvars.sys_streaming_debug_filter_min_time))
					continue;
				if(sFileFilter && sFileFilter[0] && !strstr(pFileRequest->m_strFileName.c_str(), sFileFilter))
					continue;

				const char *sFlags = (pFileRequest->m_eMediaType == eStreamSourceTypeHDD) ? "HDD" : ((pFileRequest->m_eMediaType == eStreamSourceTypeMemory) ? "mem" : "DVD");
				const char *sPriority = "";
				switch (pFileRequest->m_ePriority)
				{
				case estpUrgent:
					sPriority = "Urgent";
					break;
				case estpNormal:
					sPriority = "Normal";
					break;
				case estpIdle:
					sPriority = "  Idle";
					break;
				}

				string str;
				str.Format( "[N%6d] [%+8d] [%8d] [%6.2f ms] (%5d|%5d) [%5.3fs] <%3d> <%s> <%s> <%s> %s:",
					pFileRequest->m_nReadCounter,
					pFileRequest->m_nReadHeadOffsetKB,
					pFileRequest->m_nDiskOffset >> 10,
					pFileRequest->m_readTime.GetMilliSeconds(),
					pFileRequest->m_nSizeOnMedia/1024,
					((pFileRequest->m_nRequestedSize)?pFileRequest->m_nRequestedSize:pFileRequest->m_nFileSize)/1024,
					pFileRequest->m_completionTime*0.001,
					pFileRequest->m_nTimeGroup,
					sPriority,
					sFlags,
					pFileRequest->m_pakFile.c_str(),
					pFileRequest->m_strFileName.c_str() );

				WriteToStreamingLog( str.c_str() );
			}
		}

#endif
		return;
	}


	{
		CryAutoCriticalSection lock(m_csStats);
		tempRequests = m_statsRequestList;

		size_t nMaxRequests = g_cvars.sys_streaming_debug_filter_min_time ? 1000 : 100;
		if (m_statsRequestList.size() > nMaxRequests)
			m_statsRequestList.resize(nMaxRequests);
	}

	std::vector<CAsyncIOFileRequest_AutoPtr> &requests = tempRequests;

#ifndef _RELEASE
	stack_string temp;
	float tx = 0;
	float ty = 30;
	float ystep = 12.0f;
	float xColumn = 80;
	ColorF clText(0,1,1,1);

	//take overscan borders into account
	Vec2 overscanBorders = *(Vec2*)gEnv->pRenderer->EF_Query(EFQ_OverscanBorders);
	tx += ((float)gEnv->pRenderer->GetWidth()) * overscanBorders.x;
	ty += ((float)gEnv->pRenderer->GetHeight()) * overscanBorders.y;

	IStreamEngine::SStatistics &stats = gEnv->pSystem->GetStreamEngine()->GetStreamingStatistics();

	const char* sMediaType = m_bStreamDataOnHDD ? "HDD" : "DVD";
	const char *sStatus = (m_bStreamingStatsPaused) ? "Paused" : "";
	DrawText( tx,ty+=ystep,clText, "Streaming IO: %.2f|%.2fMB/s, ACT: %3dmsec, Unzip: %.2fMB/s, Jobs:%5d (%4d) %s %s", 
		(float)stats.nTotalCurrentReadBandwidth / (1024*1024), (float)stats.nTotalSessionReadBandwidth / (1024*1024),					
		(uint32)stats.fAverageCompletionTime, (float)stats.nDecompressBandwidth / (1024*1024),
		(uint32)stats.nTotalStreamingRequestCount, (uint32)(stats.nTotalRequestCount - stats.nTotalStreamingRequestCount), 
		sMediaType, sStatus );

	DrawText( tx,ty+=ystep,clText, "\t Request: Active:%2d (%2.1fMB) Decompress:%2d Async:%2d Finished:%2d Temp Pool Max:%2.1fMB", stats.nOpenRequestCount,
		(float)stats.nPendingReadBytes/(1024*1024),	stats.nCurrentDecompressCount, stats.nCurrentAsyncCount, stats.nCurrentFinishedCount,
		(float)stats.nMaxTempMemory/(1024*1024));

	ty+=ystep;

	// HDD stats
	if (stats.hddInfo.nTotalRequestCount > 0)
	{
		DrawText( tx,ty+=ystep,clText, "HDD : Request: %3d|%5d (%4d MB|%3d KB) - BW: %1.2f|%1.2f Mb/s (Eff: %2.1f|%2.1f Mb/s) \n", 					 
			stats.hddInfo.nRequestCount, stats.hddInfo.nTotalRequestCount, (uint32)(stats.hddInfo.nTotalBytesRead / (1024*1024)),
			(uint32)(stats.hddInfo.nTotalBytesRead / (1024*stats.hddInfo.nTotalRequestCount)),
			(float)stats.hddInfo.nCurrentReadBandwidth / (1024*1024), (float)stats.hddInfo.nSessionReadBandwidth / (1024*1024),
			(float)stats.hddInfo.nActualReadBandwidth / (1024*1024), (float)stats.hddInfo.nAverageActualReadBandwidth / (1024*1024));
		DrawText( tx,ty+=ystep,clText, "\t  Seek: %1.2f GB - Active: %2.1f%%(%2.1f%%)",	 
			(float)stats.hddInfo.nAverageSeekOffset / (1024*1024),
			stats.hddInfo.fActiveDuringLastSecond, stats.hddInfo.fAverageActiveTime);
	}
	// Optical stats
	if (stats.discInfo.nTotalRequestCount > 0)
	{
		DrawText( tx,ty+=ystep,clText, "Disc: Request: %3d|%5d (%4d MB|%3d KB) - BW: %1.2f|%1.2f Mb/s (Eff: %2.1f|%2.1f Mb/s) \n", 					 
			stats.discInfo.nRequestCount, stats.discInfo.nTotalRequestCount, (uint32)(stats.discInfo.nTotalBytesRead / (1024*1024)),
			(uint32)(stats.discInfo.nTotalBytesRead / (1024*stats.discInfo.nTotalRequestCount)),
			(float)stats.discInfo.nCurrentReadBandwidth / (1024*1024), (float)stats.discInfo.nSessionReadBandwidth / (1024*1024),
			(float)stats.discInfo.nActualReadBandwidth / (1024*1024), (float)stats.discInfo.nAverageActualReadBandwidth / (1024*1024));
		DrawText( tx,ty+=ystep,clText, "\t  Seek: %1.2f GB - Active: %2.1f%%(%2.1f%%)",	 
			(float)stats.discInfo.nAverageSeekOffset / (1024*1024),
			stats.discInfo.fActiveDuringLastSecond, stats.discInfo.fAverageActiveTime);
	}
	DrawText( tx,ty+=ystep,clText, "Mem : Request: %3d|%5d (%4d MB)", 					 
		stats.memoryInfo.nRequestCount, stats.memoryInfo.nTotalRequestCount, (stats.memoryInfo.nTotalBytesRead / (1024*1024)));


















	
	ty+=ystep;

	for(int i=eStreamTaskTypeCount-1; i>=eStreamTaskTypeTexture; i--)
	{
		EStreamTaskType eTaskType = (EStreamTaskType)i;
		IStreamEngine::SStatistics::SRequestTypeInfo info = stats.typeInfo[eTaskType];

		if (g_cvars.sys_streaming_debug > 1 || info.nTotalRequestCount > 0)
		{
			DrawText( tx,ty+=ystep,clText, "%9s: BSize:%3dKb Read:%4dMb BW:%1.2f|%1.2f Mb/s ACT:%5dms %2d(%2.1fMB)|%5d", 
				gEnv->pSystem->GetStreamEngine()->GetStreamTaskTypeName(eTaskType), 
				(uint32)(info.nTotalReadBytes/max((uint32)1,info.nTotalStreamingRequestCount)/1024),
				(uint32)(info.nTotalReadBytes/(1024*1024)),(float)info.nCurrentReadBandwidth/(1024*1024),
				(float)info.nSessionReadBandwidth/(1024*1024), (uint32)info.fAverageCompletionTime,
				info.nOpenRequestCount, (float)info.nPendingReadBytes/(1024*1024), (uint32)info.nTotalStreamingRequestCount);
		}
	}

	if (g_cvars.sys_streaming_debug == 5)
	{
		ty+=ystep;
		ty+=ystep;

		DrawText( tx,ty+=ystep,clText,"Name | Time(s) | Size(Kb) | Read(Mb) | ReqS(Mb) | Count" ); 

		for (TExtensionInfoMap::iterator it = m_PerExtensionInfo.begin(); it != m_PerExtensionInfo.end(); ++it)
		{
			SExtensionInfo& extensionInfo = it->second;
			DrawText( tx,ty+=ystep,clText,"%4s | %7.3f | %8d | %8.3f | %8.3f | %5d", 
				it->first.c_str(), extensionInfo.m_fTotalReadTime / 1000, (uint32)(extensionInfo.m_nTotalReadSize/max((size_t)1,extensionInfo.m_nTotalRequests)/1024),
				extensionInfo.m_nTotalReadSize/(1024.0f * 1024.0f), extensionInfo.m_nTotalRequestSize/(1024.0f * 1024.0f), extensionInfo.m_nTotalRequests);
		}
	}
	else if (g_cvars.sys_streaming_debug > 1)
	{
		ty+=ystep;

		DrawText( tx,ty+=ystep,clText,"[Offset KB]" );
		DrawText( tx+xColumn,ty,clText,"[io  ms]\t(read | size) [t sec] [Grp] < Prior> <Disk> \tFilename" );

		ty += ystep;

		const char* const sFileFilter = g_cvars.sys_streaming_debug_filter_file_name->GetString();

		for(size_t i = 0,nRequests = requests.size(); i < nRequests; i++)
		{
			CAsyncIOFileRequest *pFileRequest = requests[i];

			if (g_cvars.sys_streaming_debug_filter > 0 && pFileRequest->m_eType != g_cvars.sys_streaming_debug_filter)
				continue;
			if (g_cvars.sys_streaming_debug_filter == -1 && pFileRequest->m_eMediaType == eStreamSourceTypeMemory)
				continue;
			if (g_cvars.sys_streaming_debug_filter_min_time && (pFileRequest->m_readTime.GetMilliSeconds() < (float)g_cvars.sys_streaming_debug_filter_min_time))
				continue;
			if(sFileFilter!=0 && sFileFilter[0] && !strstr(pFileRequest->m_strFileName.c_str(), sFileFilter))
				continue;

			{
				float fMillis = pFileRequest->m_readTime.GetMilliSeconds();
				const char *sFlags = "";
				switch (pFileRequest->m_eMediaType)
				{
				case eStreamSourceTypeHDD:
					sFlags = "HDD";
					break;
				case eStreamSourceTypeDisc:
					sFlags = "DVD";
					break;
				case eStreamSourceTypeMemory:
					sFlags = "MEM";
					break;
				}
				const char *sPriority = "";
				switch (pFileRequest->m_ePriority)
				{
				case estpUrgent:
					sPriority = "Urgent";
					break;
				case estpNormal:
					sPriority = "Normal";
					break;
				case estpIdle:
					sPriority = "  Idle";
					break;
				}
				uint32 nRequestedSize = (pFileRequest->m_nRequestedSize != 0) ? pFileRequest->m_nRequestedSize : pFileRequest->m_nFileSize;

				//////////////////////////////////////////////////////////////////////////
				ColorF colOffset;
				if (pFileRequest->m_nReadHeadOffsetKB >= 0)
				{
					colOffset = ColorF (0,1,0,1); // Green
					if (pFileRequest->m_nReadHeadOffsetKB > 32)
					{
						colOffset = ColorF (0.5f,1.f,0,1.f); // Cyan
					}
				}
				else
				{
					colOffset = ColorF (1,0,0,1); // Red
				}
				if (pFileRequest->m_eMediaType != eStreamSourceTypeMemory)
				{
					DrawText( tx,ty,colOffset,"[%+d]",pFileRequest->m_nReadHeadOffsetKB );
				}
				//////////////////////////////////////////////////////////////////////////

				DrawText( tx+xColumn,ty,clText,"[%6.2f]\t(%5d|%5d) [%5.2f] [%3d] <%s> <%s>\t%s",
					fMillis,pFileRequest->m_nSizeOnMedia/1024,nRequestedSize/1024,pFileRequest->m_completionTime*.001f,
					pFileRequest->m_nTimeGroup, sPriority, sFlags, pFileRequest->m_strFileName.c_str() );

				ty += ystep;
			}
		}
	}
#endif //_RELEASE
}

void CStreamEngine::ClearStatistics()
{
	m_TimeOfLastReset = gEnv->pTimer->GetAsyncTime();
	m_TimeOfLastUpdate = m_TimeOfLastReset;

	m_Statistics.hddInfo.ResetStats();
	m_Statistics.discInfo.ResetStats();

	m_PerExtensionInfo.clear();

	m_Statistics.nDecompressBandwidth = 0;
	m_Statistics.nDecompressBandwidthAverage = 0;

	m_Statistics.nTotalBytesRead = 0;
	m_Statistics.nTotalRequestCount = 0;
	m_Statistics.nTotalStreamingRequestCount = 0;

	m_Statistics.nMaxTempMemory = 0;

	m_Statistics.fAverageCompletionTime = 0;

	for (int i = 0; i < eStreamTaskTypeCount; i++) {
		m_Statistics.typeInfo[i].ResetStats();
	}
	m_Statistics.vecHeavyAssets.clear();

	for (int i = 0; i < eIOThread_Last; i++)
	{
		if (m_pThreadIO[i])
		{
			m_pThreadIO[i]->m_InMemoryStats.Reset();
			m_pThreadIO[i]->m_NotInMemoryStats.Reset();
		}
	}
}

//////////////////////////////////////////////////////////////////////////
bool CStreamEngine::OnInputEvent( const SInputEvent &event )
{
	if (g_cvars.sys_streaming_debug)
	{
		if (event.keyId == eKI_F11)
		{
			m_bStreamingStatsPaused = true;
		}
		if (event.keyId == eKI_F12)
		{
			m_bStreamingStatsPaused = false;
		}
	}
	return false;
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::OnSystemEvent( ESystemEvent event,UINT_PTR wparam,UINT_PTR lparam )
{
	switch (event)
	{
	case ESYSTEM_EVENT_GAME_POST_INIT_DONE:
		{
			// unpause the streaming engine, when init phase is done
			PauseStreaming(false, -1);
			break;
		}
	case ESYSTEM_EVENT_LEVEL_LOAD_PREPARE:
		ClearStatistics();
		WriteToStreamingLog( "*LEVEL_LOAD_PREPARE" );
		break;

	case ESYSTEM_EVENT_LEVEL_LOAD_START:
		{
			WriteToStreamingLog( "*LEVEL_LOAD_START" );
			break;
		}
	case ESYSTEM_EVENT_LEVEL_LOAD_END:
		{
			WriteToStreamingLog( "*LEVEL_LOAD_END" );
			break;
		}
	case ESYSTEM_EVENT_LEVEL_PRECACHE_START:
		{
			WriteToStreamingLog( "*LEVEL_LOAD_PRECACHE_START" );
			break;
		}
	case ESYSTEM_EVENT_LEVEL_PRECACHE_END:
		{
			WriteToStreamingLog( "*LEVEL_LOAD_PRECACHE_END" );
			break;
		}
	case ESYSTEM_EVENT_LEVEL_UNLOAD:
		{
			UpdateAndWait(true);
			CancelAll();
			ClearStatistics();
			break;
		}
	case ESYSTEM_EVENT_LEVEL_POST_UNLOAD:
		{
			UpdateAndWait(true);
			CancelAll();
			ClearStatistics();
		}
		break;
	case ESYSTEM_EVENT_SHUTDOWN:
		{
			Shutdown();
			break;
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamEngine::PauseStreaming( bool bPause,uint32 nPauseTypesBitmask )
{
	if (bPause)
	{
		m_nPausedDataTypesMask |= nPauseTypesBitmask;
	}
	else
	{
		m_nPausedDataTypesMask &= ~nPauseTypesBitmask;
		ResumePausedStreams();
	}
}
//////////////////////////////////////////////////////////////////////////
void CStreamEngine::PauseIO( bool bPause )
{
	for (int i = 0; i < eIOThread_Last; i++)
	{
		if (m_pThreadIO[i])
			m_pThreadIO[i]->Pause(bPause);
	}
}

void CStreamEngine::PushRequestToAsyncCallbackThread( CAsyncIOFileRequest *pFileRequest )
{
	int nCallbackThreads = m_asyncCallbackThreads.size();
	if (nCallbackThreads > 0)
	{
		if (nCallbackThreads > 1 && pFileRequest->m_eType == eStreamTaskTypeGeometry)
		{
			// If we have more then 1 call back threads, use this one for geometry only.
			m_asyncCallbackThreads[1]->AddRequest(pFileRequest);
		}
		else
		if (nCallbackThreads > 2 && pFileRequest->m_eType == eStreamTaskTypeTexture)
		{
			// If we have more then 1 call back threads, use this one for geometry only.
			m_asyncCallbackThreads[2]->AddRequest(pFileRequest);
		}
		else
		{
			m_asyncCallbackThreads[0]->AddRequest(pFileRequest);
		}
	}
}
//////////////////////////////////////////////////////////////////////////
